package org.csc.gateway.filter;

import com.alibaba.fastjson.JSONObject;
import io.github.bucket4j.Bucket;
import io.netty.buffer.ByteBufAllocator;
import lombok.extern.slf4j.Slf4j;
import org.csc.gateway.config.Config;
import org.csc.gateway.util.IpUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.cloud.gateway.route.Route;
import org.springframework.context.ApplicationContext;
import org.springframework.core.Ordered;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.NettyDataBufferFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.server.RequestPath;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpRequestDecorator;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.net.InetSocketAddress;
import java.net.URI;
import java.nio.CharBuffer;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;

/**
 * Only one connection receive subscriber allowed. // 订阅发布模式
 * 错误原因：
 * 实际上spring-cloud-gateway反向代理的原理是，首先读取原请求的数据，然后构造一个新的请求，将原请求的数据封装到新的请求中，
 * 然后再转发出去。然而我们在他封装之前读取了一次request body，而request body只能读取一次。因此就出现了上面的错误。
 * <p>
 * 对于上面的错误我们给出的解决方案是：// 装饰者模式
 * 读取request body的时候，我们再封装一次request，转发出去
 */
@Slf4j
@Component
public class RequestGlobalFilter implements GlobalFilter, Ordered {
    private static final String POST = "POST";
    private static final Map<String, Bucket> CACHE = new ConcurrentHashMap<>();
    private static final long NUM_TOKENS = 1;
    private static final int CAPACITY = 10; // 总共令牌数
    private static final int REFILL_TOKENS = 1; // 值越小达到请求上限后，停止请求的时间越长
    private static final long SECONDS = 10;  // 令牌刷新时间 单位秒

    @Autowired
    private Config config;

    @Override
    public int getOrder() {
        return 0;
    }

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        if (config.getJournalEnabled()) {
            ServerHttpRequest request = this.getServerHttpRequest(exchange);
            return chain.filter(exchange.mutate().request(request).build());
        }

        if (config.getCurrentLimitingEnabled()) {
            if (config.getCurrentLimitingIp()) {
                String ip = exchange.getRequest().getRemoteAddress().getAddress().getHostAddress();
                IpUtils ipUtils = new IpUtils(CAPACITY, REFILL_TOKENS, Duration.ofSeconds(SECONDS));
                Bucket bucket = CACHE.computeIfAbsent(ip, k -> ipUtils.createNewBucket());
                log.info("IP: " + ip + "，TokenBucket Available Tokens: " + bucket.getAvailableTokens());
                if (!bucket.tryConsume(NUM_TOKENS)) {
                    exchange.getResponse().setStatusCode(HttpStatus.TOO_MANY_REQUESTS);
                    return exchange.getResponse().setComplete();
                }
            }
            if (config.getCurrentLimitingUrl()) {

            }
            if (config.getCurrentLimitingUser()) {

            }
        }

        return chain.filter(exchange);
    }

    private ServerHttpRequest getServerHttpRequest(ServerWebExchange exchange) {
        Map<String, Object> map = exchange.getAttributes();
        Route route = (Route) exchange.getAttribute("org.springframework.cloud.gateway.support.ServerWebExchangeUtils.gatewayRoute");
        ApplicationContext applicationContext = exchange.getApplicationContext();
        ServerHttpRequest request = exchange.getRequest();
        HttpHeaders httpHeaders = request.getHeaders();
        String mo = request.getMethodValue();
        URI uri = request.getURI();
        RequestPath requestPath = request.getPath();
        InetSocketAddress socketAddress = request.getRemoteAddress();
        boolean a = httpHeaders.containsKey("token");
        List<String> values = httpHeaders.get("token");
        if (null != values && !values.isEmpty()) {
            log.info("values  -----> {}", values.toString());
        }
        if (POST.equalsIgnoreCase(request.getMethodValue())) {
            //判断是否为POST请求
            Flux<DataBuffer> body = request.getBody();
            //缓存读取的request body信息
            AtomicReference<String> bodyRef = new AtomicReference<>();
            //读取request body到缓存
            body.subscribe(dataBuffer -> {
                CharBuffer charBuffer = StandardCharsets.UTF_8.decode(dataBuffer.asByteBuffer());
                DataBufferUtils.release(dataBuffer);
                bodyRef.set(charBuffer.toString());
            });
            HttpMethod method = request.getMethod();
            //获取request body
            String bodyStr = bodyRef.get();
            log.info("bodyStr  -----> {}", bodyStr);
            DataBuffer bodyDataBuffer = stringBuffer(bodyStr);
            Flux<DataBuffer> bodyFlux = Flux.just(bodyDataBuffer);

            //封装我们的request
            request = new ServerHttpRequestDecorator(request) {
                @Override
                public Flux<DataBuffer> getBody() {
                    return bodyFlux;
                }
            };
        } else {
            Map<String, String> queryParansMap = request.getQueryParams().toSingleValueMap();
            if (null != queryParansMap && !queryParansMap.isEmpty()) {
                log.info("queryParans  -----> {}", JSONObject.toJSON(queryParansMap).toString());
            }
        }
        return request;
    }

    protected DataBuffer stringBuffer(String value) {
        byte[] bytes = value.getBytes(StandardCharsets.UTF_8);

        NettyDataBufferFactory nettyDataBufferFactory = new NettyDataBufferFactory(ByteBufAllocator.DEFAULT);
        DataBuffer buffer = nettyDataBufferFactory.allocateBuffer(bytes.length);
        buffer.write(bytes);
        return buffer;
    }

}
